/**
 * Copyright (C), 2011-2016
 * File Name: RefreshableLocalCache.java
 * Encoding: UTF-8
 * Date: 2012-11-5
 * History:
 */
package com.swak.frame.job;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.slf4j.Logger;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;

import com.alibaba.fastjson.JSON;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.swak.frame.cache.CacheProxy;
import com.swak.frame.exception.SwakRuntimeException;
import com.swak.frame.logger.SwakLoggerFactory;
import com.swak.frame.timer.HashedWheelTimer;
import com.swak.frame.timer.Timeout;
import com.swak.frame.timer.Timer;
import com.swak.frame.timer.TimerTask;
import com.swak.frame.timer.WheelTimerHolder;
import com.swak.frame.util.DateTimeUtils;
import com.swak.frame.util.ObjectUtil;

/**
 * 可自动刷新的本地缓存 ClassName: RefreshableLocalCache.java
 * 
 * @author colley.ma
 * @date 2021年3月17日 上午11:26:31
 */
public abstract class RefreshableStateCache<K, V>
    implements RefreshableCache<K, V>, TimerTask, InitializingBean {
    private static final int MAX_RETRIES = 100;
    protected Logger logger = SwakLoggerFactory.getLogger(this.getClass());

    /** 重试次数,小于5次,reload数据 */
    private AtomicInteger retriesTimer = new AtomicInteger(0);

    /** 更新时设置为不可读，获取数据时 睡眠或自旋 **/
    private AtomicBoolean read = new AtomicBoolean(true);
    private AtomicBoolean ayncRunning = new AtomicBoolean(false);

    @Autowired(required = false)
    protected CacheProxy cacheProxy;

    /** 主缓存 */
    private Cache<K, V> segment;
    private HashedWheelTimer hashedWheelTimer;

    private boolean remoteCache = false;

    /** 缓存中允许的最大条目数。默认值为1000 **/
    private int maxSize = 1000;

    /** 分布式缓存时间 **/
    private long distributedPeriod = 600;

    private RefreshRate refreshRate;

    /**
     * 
     * @return the maxSize
     */
    public int getMaxSize() {
        return maxSize;
    }

    /**
     * 
     * @param maxSize the maxSize to set
     */
    public void setMaxSize(int maxSize) {
        this.maxSize = maxSize;
    }

    @Override
    public void refresh() {
        try {
            logger.warn("Start refresh cache at {},taskName:{}", DateTimeUtils.sysTimeStr(),
                this.getClass().getSimpleName());

            long startTime = System.currentTimeMillis();
            List<V> result = this.loadDataFrom();
            if (!isValid(result)) {
                logger.error("Data validation failed,cacheName:{},result:{}", cacheName(),
                    JSON.toJSONString(result));
            }
            Map<K, V> bizData = this.handler(result);
            if (MapUtils.isEmpty(bizData)) {
                logger.warn("refresh cache successfully,taskName:{},time:{} bizData is empty!",
                    new Object[] {this.getClass().getSimpleName(), DateTimeUtils.sysTimeStr()});

                // 重试次数+1
                int retriesNum = retriesTimer.getAndIncrement();

                // 大于重试的次数阀值，清掉缓存信息
                if ((retriesNum > MAX_RETRIES) && !isEmpty()) {
                    segment.invalidateAll();
                }

                return;
            }

            // load data isNotEmpty
            // 重置重试次数为0
            retriesTimer.set(0);
            // 设置不可读
            read.set(false);
            // 执行缓存操作
            segment.invalidateAll();
            segment.putAll(bizData);
            // 返回为可读
            read.set(true);

            long endTime = System.currentTimeMillis();
            logger.warn("relaod cache successfully,taskName:{},time:{},cost:{} ms",
                new Object[] {this.getClass().getSimpleName(), DateTimeUtils.sysTimeStr(),
                    (endTime - startTime)});
        } finally {
            this.ayncRunning.compareAndSet(true, false);
        }
    }

    public List<V> loadDataFrom() {
        if (cacheProxy == null && remoteCache) {
            logger.warn("remoteCache set true,but cacheProxy is null call loadData,cacheName:{}",
                cacheName());
            return loadData();
        }

        if (!remoteCache) {
            logger.warn("remoteCache set false,call loadData,cacheName:{}", cacheName());
            return loadData();
        }

        String distributedKey = cacheProxy.format(distributedKey());
        List<V> result = cacheProxy.getObject(distributedKey);
        if (CollectionUtils.isNotEmpty(result)) {
            return result;
        }

        result = loadData();
        if (CollectionUtils.isNotEmpty(result)) {
            // 判断缓存时间存在
            if (!cacheProxy.exists(distributedKey)) {
                cacheProxy.setObject(distributedKey, result,
                    refreshRate.getDistributedRate(distributedPeriod), TimeUnit.SECONDS);
            }
        }
        return result;
    }

    @Override
    public void run(Timeout timeout) throws Exception {
        try {
            this.refresh();
        } catch (Exception e) {
            logger.error("run loadData job error", e);
        }
        reput(timeout, refreshRate.getRate());
    }

    private void reput(Timeout timeout, Long tick) {
        if ((timeout == null) || (tick == null)) {
            throw new IllegalArgumentException();
        }

        Timer timer = timeout.timer();

        if (timer.isStop() || timeout.isCancelled()) {
            return;
        }

        timer.newTimeout(timeout.task(), tick, TimeUnit.SECONDS);
    }

    /**
     * 读取缓存中某个条目
     *
     * @param key
     *
     * @return
     */
    public V get(K key) {
        // 非可读，睡眠10ms
        if (!this.read.get()) {
            try {
                Thread.sleep(10L);
            } catch (InterruptedException e) {
                // igore
            }
        }

        V result = segment.getIfPresent(key);

        // 不为空直接返回
        if (!Objects.isNull(result)) {
            return ObjectUtil.cloneIfPossible(result);
        }

        // 为空 reload 首次直接reload 其它定时1s之后加载
        int retriesTime = retriesTimer.get();

        if ((retriesTime < MAX_RETRIES)) {
            if (isEmpty() && !this.ayncRunning.get()) {
                logger.warn(
                    "data is empty ayncWheelTimer 1s relaod data start, retriesCount:{}......",
                    retriesTime);
                this.ayncRunning.set(true);

                if (retriesTime <= 1) {
                    refresh(); // 重试1次 直接出发reload 尽快获取数据
                    result = segment.getIfPresent(key);
                } else {
                    // 异步去执行
                    WheelTimerHolder.refreshableWheel().newTimeout(new AyncTimeTask<K, V>(this), 1,
                        TimeUnit.SECONDS);
                }
            }
        }

        return result;
    }

    /**
     * 没有取到设置
     * 
     * @param key
     * @param mappingFunction
     * @return
     */
    public V get(K key, Function<? super K, ? extends V> mappingFunction) {
        V result = get(key);
        if (result == null && mappingFunction != null) {
            result = mappingFunction.apply(key);
            if (result != null) {
                segment.put(key, result);
            }
        }
        return result;
    }

    /**
     * @Desc 读取缓存中的所有对象
     * @param @return
     * @return Collection<V>
     */
    public Collection<V> readAllEntities() {
        return segment.asMap().values();
    }

    @Override
    public Map<?, ?> allEntities() {
        return segment.asMap();
    }

    protected boolean isEmpty() {
        return segment.estimatedSize() <= 0;
    }

    /**
     * 清理缓存。<br>
     * 默认实现为清除缓存中的所有数据。子类可以根据需要覆盖它
     */
    @Override
    public synchronized void clear() {
        logger.warn("Start clear cache at {}", DateTimeUtils.sysTimeStr());
        long entriesCount = segment.estimatedSize();
        segment.invalidateAll();
        if (cacheProxy != null && remoteCache) {
            String distributedKey = cacheProxy.format(distributedKey());
            logger.warn("last localcache jimdb distributedKey :{}", distributedKey);
            cacheProxy.delete(distributedKey);
        }
        logger.warn("Clear cache successfully,entries count :{}", entriesCount);
    }

    public void setHashedWheelTimer(HashedWheelTimer hashedWheelTimer) {
        this.hashedWheelTimer = hashedWheelTimer;
    }

    /**
     * @Desc 启动定时刷新缓存任务
     * @param
     * @return void
     */
    @Override
    public void startJob() {
        if (refreshRate == null) {
            throw new SwakRuntimeException(
                this.getClass().getName() + " swakJob config is illegal cron is null,fixedRate<=0");
        }
        long delay = ThreadLocalRandom.current().nextLong(20, 50);
        logger.warn("Start {} refreshCache start for delay {} seconds ",
            this.getClass().getSimpleName(), delay);
        this.hashedWheelTimer.newTimeout(this, delay, TimeUnit.SECONDS);
        logger.warn("Start {} start successfully", this.getClass().getSimpleName());
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        this.segment = Caffeine.newBuilder().maximumSize(maxSize).build();
        if (this.hashedWheelTimer == null) {
            this.hashedWheelTimer = WheelTimerHolder.refreshableWheel();
        }
    }

    public static class AyncTimeTask<K, V> implements TimerTask {
        private RefreshableCache<K, V> timerTask;

        public AyncTimeTask(RefreshableCache<K, V> timerTask) {
            this.timerTask = timerTask;
        }

        @Override
        public void run(Timeout timeout) throws Exception {
            if (timerTask != null) {
                timerTask.refresh();
            }
        }
    }

    @Override
    public String cacheName() {
        return this.getClass().getName();
    }

    public void setRemoteCache(boolean remoteCache) {
        this.remoteCache = remoteCache;
    }

    public void setDistributedPeriod(long distributedPeriod) {
        this.distributedPeriod = distributedPeriod;
    }

    @Override
    public void setRefreshRate(RefreshRate refreshRate) {
        if (this.refreshRate == null) {
            this.refreshRate = refreshRate;
        }
    }
}
